package com.bwie.service.impl;

import com.alibaba.fastjson.JSON;
import com.bwie.domain.TbLog;
import com.bwie.mapper.TbLogMapper;
import com.bwie.service.TbLogService;
import com.bwie.utils.R;
import com.bwie.vo.MessageVo;
import com.bwie.vo.PageInfoVo;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Service;

/**
 * @author 军哥
 * @version 1.0
 * @description: TODO
 * @date 2024/10/22 15:32
 */

@Service
public class TbLogServiceImpl implements TbLogService {

    @Autowired
    TbLogMapper tbLogMapper;

    @Autowired
    StringRedisTemplate stringRedisTemplate;

    @Override
    public R getLogList(PageInfoVo pageInfoVo) {
        return null;
    }

    @KafkaListener(topics = "topic-log")
    public void recvLogMessage(String message, Acknowledgment acknowledgment) {
        System.out.println("接收到kafka日志消息：" + message);
        //-- 解析消息
        MessageVo messageVo = JSON.parseObject(message, MessageVo.class);

        // 获取消息的唯一ID，进行判断是否重复
        String msgId = messageVo.getMsgId();
        Boolean hasKey = stringRedisTemplate.hasKey(msgId);
        if(!hasKey) {
            System.out.println("消息重复消费：" + msgId);
            // 删除消息
            acknowledgment.acknowledge();
            return;
        }

        //-- 消费消息
        String msgBody = messageVo.getMsgBody();
        TbLog tbLog = JSON.parseObject(msgBody, TbLog.class);
        tbLogMapper.insert(tbLog);

        //-- 手动确认消息，防止消息丢失
        acknowledgment.acknowledge();
        // 删除消息的唯一ID
        stringRedisTemplate.delete(msgId);

        System.out.println("日志消息消费完毕");
    }
}
